package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.aggregations.LongTermsBucket;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.NestedQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch._types.query_dsl.TermsQueryField;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import co.elastic.clients.json.JsonData;
import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.CategoryFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.common.util.PinYinUtils;
import com.atguigu.tingshu.model.album.*;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.search.repository.AlbumIndexRepository;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.repository.SuggestIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;

import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class SearchServiceImpl implements SearchService {

    @Autowired
    private SuggestIndexRepository suggestIndexRepository;

    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    private CategoryFeignClient categoryFeignClient;

    @Autowired
    private UserInfoFeignClient userInfoFeignClient;

    @Autowired
    private AlbumIndexRepository albumIndexRepository;

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    @Override
    public List<AlbumInfoIndex> findRankingList(Long category1Id, String dimension) {
        //  hget key field
        //  定义缓存的key  {“id”:443,"albumTitle":} ---> JSON.parseObject(str,Class.class);
        String rangKey = RedisConstant.RANKING_KEY_PREFIX+category1Id;
        //  List<AlbumInfoIndex> albumInfoIndexList = (List<AlbumInfoIndex>) this.redisTemplate.boundHashOps(rangKey).get(dimension);
        List<AlbumInfoIndex> albumInfoIndexList = (List<AlbumInfoIndex>) this.redisTemplate.opsForHash().get(rangKey, dimension);
        //  返回数据
        return albumInfoIndexList;
    }

    @Override
    public void updateLatelyAlbumRanking() {
        //  先获取到所有一级分类数据
        Result<List<BaseCategory1>> baseCategory1Result = categoryFeignClient.findAllCategory1();
        //  获取集合
        Assert.notNull(baseCategory1Result,"一级分类结果集为空");
        List<BaseCategory1> baseCategory1List = baseCategory1Result.getData();
        Assert.notNull(baseCategory1List,"一级分类集合为空");
        //  循环遍历
        for (BaseCategory1 baseCategory1 : baseCategory1List) {
            //  baseCategory1.getId()
            //  Hash数据结构： hset key field value;  key=category1Id field = 热度/播放量/订阅量/购买量/评论数 value=热度data/播放量data;
            String[] rankingDimensionArray = new String[]{"hotScore", "playStatNum", "subscribeStatNum", "buyStatNum", "commentStatNum"};
            //  排行榜数据从哪里来的? es 中!
            for (String ranging : rankingDimensionArray) {
                //  执行dsl语句 获取到结果集.
                SearchResponse<AlbumInfoIndex> response = null;
                try {
                    response = elasticsearchClient.search(f -> f.index("albuminfo")
                            .query(q -> q.term(t -> t.field("category1Id").value(baseCategory1.getId())))
                            .sort(s -> s.field(d -> d.field(ranging).order(SortOrder.Desc)))
                            .size(10), AlbumInfoIndex.class);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                //  获取到当前执行结果集，将数据存储到缓存：
                List<AlbumInfoIndex> albumInfoIndexList = response.hits().hits().stream().map(Hit::source).collect(Collectors.toList());
                //  使用hash 数据结构 hset key field value;
                String rangKey = RedisConstant.RANKING_KEY_PREFIX+baseCategory1.getId();
                this.redisTemplate.boundHashOps(rangKey).put(ranging,albumInfoIndexList);
                //  this.redisTemplate.opsForHash().put(rangKey,ranging,albumInfoIndexList);
            }
        }
    }

    @Override
    public List<String> completeSuggest(String keyword) {
        //  查询数据：动态生成dsl 语句.
        SearchRequest.Builder request = new SearchRequest.Builder();
        //  动态生成的DSL语句.
        request.index("suggestinfo").suggest(
                s->s.suggesters("suggestionKeyword",f->f.prefix(keyword).completion(
                        c->c.field("keyword").skipDuplicates(true).size(10).fuzzy(
                                z->z.fuzziness("auto")
                        )
                ))
                        .suggesters("suggestionkeywordPinyin",f->f.prefix(keyword).completion(
                                c->c.field("keywordPinyin").skipDuplicates(true).size(10).fuzzy(
                                        z->z.fuzziness("auto")
                                )
                        ))
                        .suggesters("suggestionkeywordSequence",f->f.prefix(keyword).completion(
                                c->c.field("keywordSequence").skipDuplicates(true).size(10).fuzzy(
                                        z->z.fuzziness("auto")
                                )
                        ))
        );
        //  查询数据：
        SearchResponse<SuggestIndex> response = null;
        try {
            response = elasticsearchClient.search(request.build(), SuggestIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        //  从提词库中获取结果集.
        Set<String> titleSet = new HashSet<>();
        titleSet.addAll(this.parseResultData(response,"suggestionKeyword"));
        titleSet.addAll(this.parseResultData(response,"suggestionkeywordPinyin"));
        titleSet.addAll(this.parseResultData(response,"suggestionkeywordSequence"));
        //  如果不满10条数据，查询提词库数据进行补救！
        if (titleSet.size()<10){
            //  直接查询
            SearchRequest.Builder requestBuilder = new SearchRequest.Builder();
            requestBuilder.index("suggestinfo").query(f->f.match(m->m.field("title").query(keyword)));
            //  获取返回结果集
            SearchResponse<SuggestIndex> searchResponse = null;
            try {
                searchResponse = elasticsearchClient.search(requestBuilder.build(), SuggestIndex.class);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            searchResponse.hits().hits().stream().forEach(hit->{
                //  获取数据
                SuggestIndex suggestIndex = hit.source();
                //  获取专辑标题
                String title = suggestIndex.getTitle();
                //  添加到集合中
                titleSet.add(title);
                //  判断
                if (titleSet.size()==10){
                    return ;
                }
            });
        }
        //  返回集合数据
        return new ArrayList<>(titleSet);
    }

    /**
     * 获取提词数据
     * @param response
     * @param keyword
     * @return
     */
    private List<String> parseResultData(SearchResponse<SuggestIndex> response, String keyword) {
        //  声明一个集合
        //  List<String> list = new ArrayList<>();
        //  从提词库中获取数据.
        List<Suggestion<SuggestIndex>> suggestionList = response.suggest().get(keyword);
        //  map 映射关系.
//        suggestionList.stream().forEach(suggest->{
//            suggest.completion().options().stream().forEach(item->{
//                //  获取提词对象
//                SuggestIndex suggestIndex = item.source();
//                list.add(suggestIndex.getTitle());
//            });
//            return suggest.completion().options().stream().map(item -> item.source().getTitle()).collect(Collectors.toList());
//            return list;
//       });
        for (Suggestion<SuggestIndex> suggestIndexSuggestion : suggestionList) {
            //  获取提词数据
            return suggestIndexSuggestion.completion().options().stream().map(item -> item.source().getTitle()).collect(Collectors.toList());
        }
        //  提词集合。
        return null;
    }

    @Override
    public List<HashMap<String, Object>> channel(Long category1Id) {
        //  声明集合
        List<HashMap<String, Object>> list = new ArrayList<>();
        //  根据一级分类Id 获取三级数据.
        Result<List<BaseCategory3>> baseCategory3ListResult = categoryFeignClient.findTopBaseCategory3(category1Id);
        Assert.notNull(baseCategory3ListResult, "三级分类数据结果集为空");
        List<BaseCategory3> baseCategory3List = baseCategory3ListResult.getData();
        Assert.notNull(baseCategory3List, "三级分类数据集合为空");
        //  将baseCategory3List 转换为 map 集合 key = category3Id, value = BaseCategory3;
        Map<Long, BaseCategory3> category3IdToMap = baseCategory3List.stream().collect(Collectors.toMap(BaseCategory3::getId, baseCategory3 -> baseCategory3));
        //  baseCategory3List ---> 根据一级分类Id查询置顶的三级分类数据.
        //  根据三级分类 Id 获取专辑数据.
        List<Long> category3IdList = baseCategory3List.stream().map(BaseCategory3::getId).collect(Collectors.toList());
        //  List<FieldValue>
        //        List<FieldValue> fieldValueList = category3IdList.stream().map(category3Id -> {
        //            FieldValue.Builder fieldValue = new FieldValue.Builder();
        //            fieldValue.longValue(category3Id);
        //            return fieldValue.build();
        //        }).collect(Collectors.toList());
        //  es 如何将数据存储到 List<FieldValue>
        List<FieldValue> fieldValueList = category3IdList.stream().map(category3Id -> new FieldValue.Builder().longValue(category3Id).build()).collect(Collectors.toList());
        //  获取request.Builder 对象
        SearchRequest.Builder request = new SearchRequest.Builder();
        request.index("albuminfo").query(q -> q.terms(t -> t.field("category3Id").terms(new TermsQueryField.Builder().value(fieldValueList).build())));
        //  根据三级分类Id 进行分组获取签6条数据记录并按照热度或播放量进行降序排列.
        request.aggregations("groupByCategory3IdAgg", a -> a.terms(
                        t -> t.field("category3Id").size(10))
                .aggregations("topTenHotScoreAgg", t -> t.topHits(h -> h.size(6)
                        .sort(s -> s.field(f -> f.field("hotScore").order(SortOrder.Desc))))));
        //  声明一个响应对象
        SearchResponse<AlbumInfoIndex> response = null;
        try {
            response = elasticsearchClient.search(request.build(), AlbumInfoIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        //  获取专辑集合：应该获取按照分类Id分组之后的聚合数据。
        Aggregate groupByCategory3IdAgg = response.aggregations().get("groupByCategory3IdAgg");
        if (null != groupByCategory3IdAgg){
            List<LongTermsBucket> bucketList = groupByCategory3IdAgg.lterms().buckets().array();
            if (!CollectionUtils.isEmpty(bucketList)){
                  list = bucketList.stream().map(bucket -> {
                    //  获取到三级分类Id
                    long category3Id = bucket.key();
                    //  创建Map 集合
                    HashMap<String, Object> map = new HashMap<>();
                    //  获取子聚合数据
                    Aggregate topTenHotScoreAgg = bucket.aggregations().get("topTenHotScoreAgg");
                    Assert.notNull(topTenHotScoreAgg, "子聚合数据为空.");
                    List<Hit<JsonData>> hits = topTenHotScoreAgg.topHits().hits().hits();
                    Assert.notNull(hits, "集合数据为空.");
                    //  获取数据
                    List<AlbumInfoIndex> albumInfoIndexList = hits.stream().map(hit -> {
                        String str = hit.source().toString();
                        AlbumInfoIndex albumInfoIndex = JSON.parseObject(str, AlbumInfoIndex.class);
                        return albumInfoIndex;
                    }).collect(Collectors.toList());

                    //                    List<JsonData> collect = hits.stream().map(Hit::source).collect(Collectors.toList());
                    //	根据三级分类Id 找到三级分类对象
                    map.put("baseCategory3", category3IdToMap.get(category3Id));
                    //	专辑的集合数据
                    map.put("list", albumInfoIndexList);
                    return map;
                }).collect(Collectors.toList());
            }
        }
        //  返回数据
        return list;
    }

    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery albumIndexQuery) {
        //  创建对象
        AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();
        //  1.  封装一个dsl语句 根据用户输入的检索条件。
        SearchRequest searchRequest = this.buildQueryDsl(albumIndexQuery);
        //  2.  调用查询方法
        SearchResponse<AlbumInfoIndex> response = null;
        try {
            response = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        //  3.  数据封装 response--->albumSearchResponseVo
        //  在下面这个方法中 给 List<AlbumInfoIndexVo> list 赋值
        albumSearchResponseVo = this.parseResultData(response);
        //  本质给实体类属性赋值：
        //  List<AlbumInfoIndexVo> list = new ArrayList<>();
        //  Long total; 总记录数
        //  Integer pageSize; 每页显示的内容
        //  Integer pageNo; 当前页面
        //  Long totalPages;
        long total = response.hits().total().value();
        albumSearchResponseVo.setTotal(total);
        albumSearchResponseVo.setPageSize(albumIndexQuery.getPageSize());
        albumSearchResponseVo.setPageNo(albumIndexQuery.getPageNo());
        //  10 3 4 | 9 3 3
        //  Long totalPages = total%albumIndexQuery.getPageSize()==0?total/albumIndexQuery.getPageSize():total/albumIndexQuery.getPageSize()+1;
        Long totalPages = (total + albumIndexQuery.getPageSize() - 1) / albumIndexQuery.getPageSize();
        albumSearchResponseVo.setTotalPages(totalPages);
        //  返回对象
        return albumSearchResponseVo;
    }

    /**
     * 获取查询请求对象
     *
     * @param albumIndexQuery
     * @return
     */
    private SearchRequest buildQueryDsl(AlbumIndexQuery albumIndexQuery) {
        //  创建一个 SearchRequest
        SearchRequest.Builder request = new SearchRequest.Builder();
        BoolQuery.Builder bool = new BoolQuery.Builder();
        //  判断关键词是否为空！
        if (!StringUtils.isEmpty(albumIndexQuery.getKeyword())) {
            //  { query - bool - should - match -{key-value}} 复制小括号，写死右箭头，落地大括号
            //  request.query(q->q.bool(s->s.should(m->m.match(f->f.field("albumTitle").query(albumIndexQuery.getKeyword())))));
            //  request.query(q->q.bool(s->s.should(m->m.match(f->f.field("albumIntro").query(albumIndexQuery.getKeyword())))));
            bool.should(m -> m.match(f -> f.field("albumTitle").query(albumIndexQuery.getKeyword())));
            bool.should(m -> m.match(f -> f.field("albumIntro").query(albumIndexQuery.getKeyword())));
            //  生成高亮的dsl 语句. 设置高亮规则
            request.highlight(f -> f.fields("albumTitle", h -> h.preTags(Arrays.asList("<span style=color:red>")).postTags(Arrays.asList("</span>"))));
            //  利用可变参数的特性：可变参数必须放在最后，那么这个参数可有可无!
            //  request.highlight(f->f.fields("albumTitle",h->h.preTags("<span style=color:red>").postTags("</span>")));
        }
        //  根据分类Id
        if (!StringUtils.isEmpty(albumIndexQuery.getCategory1Id())) {
            bool.filter(f -> f.term(t -> t.field("category1Id").value(albumIndexQuery.getCategory1Id())));
        }
        if (!StringUtils.isEmpty(albumIndexQuery.getCategory2Id())) {
            bool.filter(f -> f.term(t -> t.field("category2Id").value(albumIndexQuery.getCategory2Id())));
        }
        if (!StringUtils.isEmpty(albumIndexQuery.getCategory3Id())) {
            bool.filter(f -> f.term(t -> t.field("category3Id").value(albumIndexQuery.getCategory3Id())));
        }

        //  属性Id 过滤 先获取数据 属性id:属性值id&属性id:属性值id
        List<String> attributeList = albumIndexQuery.getAttributeList();
        //  判断
        if (!CollectionUtils.isEmpty(attributeList)) {
            //  循环遍历
            for (String attr : attributeList) {
                //  分割数据 属性id=split[0]  属性值id=split[1]
                String[] split = attr.split(":");
                //  判断
                if (null != split && split.length == 2) {
                    //  构建nested
                    Query nested = NestedQuery.of(q -> q.path("attributeValueIndexList")
                            .query(b -> b.bool(f -> f.filter(t -> t.term(s -> s.field("attributeValueIndexList.attributeId").value(split[0])))
                                            .filter(t -> t.term(s -> s.field("attributeValueIndexList.valueId").value(split[1])))
                                    )
                            )
                    )._toQuery();
                    //  bool -- filter;
                    bool.filter(nested);
                }
            }
        }

        //  分页 10 3  第一页 0,3  第二页 3,3 第三页 6,3
        Integer from = (albumIndexQuery.getPageNo() - 1) * albumIndexQuery.getPageSize();
        request.from(from); // 从第几条数据开始
        request.size(albumIndexQuery.getPageSize()); // 每页显示的条数
        //  排序
        //  request.sort(s->s.field(f->f.field("id").order(SortOrder.Desc)));
        //  综合排序[1:desc] hotScore 播放量[2:desc] playStatNum  发布时间[3:desc] createTime asc:升序 desc:降序
        String order = albumIndexQuery.getOrder();
        //  定义一个变量来存储排序字段
        String orderFiled = "";
        //  判断
        if (!StringUtils.isEmpty(order)) {
            //  order = 1:desc or 2:desc or 3:desc
            //  根据1,2,3 判断传递的字段是什么!
            String[] split = order.split(":");
            //  判断数组
            if (null != split && split.length > 0) {
                switch (split[0]) {
                    case "1":
                        orderFiled = "hotScore";
                        break;
                    case "2":
                        orderFiled = "playStatNum";
                        break;
                    case "3":
                        orderFiled = "createTime";
                        break;
                }
                //  设置排序规则
                String finalOrderFiled = orderFiled;
                request.sort(s -> s.field(f -> f.field(finalOrderFiled).order("asc".equals(split[1]) ? SortOrder.Asc : SortOrder.Desc)));
            }
        } else {
            //    默认排序规则
            request.sort(s -> s.field(f -> f.field("hotScore").order(SortOrder.Desc)));
        }
        //  其他的设置：
        request.source(s -> s.filter(f -> f.excludes(Arrays.asList("category1Id", "category2Id", "category3Id", "attributeValueIndexList"))));
        //  {query--bool}
        request.query(bool.build()._toQuery());
        //  设置索引库.
        SearchRequest searchRequest = request.index("albuminfo").build();
        System.out.println("DSL:\t" + searchRequest.toString());
        //  返回查询对象
        return searchRequest;
    }

    private AlbumSearchResponseVo parseResultData(SearchResponse<AlbumInfoIndex> response) {
        //  创建对象
        AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();
        //  赋值：List<AlbumInfoIndexVo> list = new ArrayList<>();
        List<Hit<AlbumInfoIndex>> hits = response.hits().hits();
        //  判断集合不为空
        if (!CollectionUtils.isEmpty(hits)) {
            //  使用map 来获取数据对象。
            List<AlbumInfoIndexVo> list = hits.stream().map(hit -> {
                //  获取专辑对象信息
                AlbumInfoIndex albumInfoIndex = hit.source();
                //  创建一个vo对象
                AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
                //  属性拷贝：
                BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
                //  判断用户是否通过关键词进行了检索
                if (null != hit.highlight().get("albumTitle")) {
                    //  获取高亮字段
                    String albumTitle = hit.highlight().get("albumTitle").get(0);
                    //  赋值高亮字段
                    albumInfoIndexVo.setAlbumTitle(albumTitle);
                }
                //  返回数据
                return albumInfoIndexVo;
            }).collect(Collectors.toList());
            //  赋值数据.
            albumSearchResponseVo.setList(list);
        }
        //  返回数据
        return albumSearchResponseVo;
    }

    @Override
    public void lowerAlbum(Long albumId) {
        //  删除数据
        albumIndexRepository.deleteById(albumId);
    }

    @Override
    public void upperAlbum(Long albumId) {
        //  创建对象
        //  8.5.0
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
        //  赋值 使用有返回值！ albumInfo 的其他属性需要作为后续的参数传递！
        CompletableFuture<AlbumInfo> albumInfoCompletableFuture = CompletableFuture.supplyAsync(() -> {
            //  获取专辑对象信息
            Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfo(albumId);
            Assert.notNull(albumInfoResult, "远程调用专辑返回值不能为空.");
            AlbumInfo albumInfo = albumInfoResult.getData();
            Assert.notNull(albumInfo, "专辑不能为空.");
            BeanUtils.copyProperties(albumInfo, albumInfoIndex);
            return albumInfo;
        }, threadPoolExecutor);

        //  设置主播名称
        CompletableFuture<Void> userInfoCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            Result<UserInfoVo> userInfoVoResult = userInfoFeignClient.getUserInfoVo(albumInfo.getUserId());
            Assert.notNull(userInfoVoResult, "远程调用用户返回值不能为空.");
            UserInfoVo userInfoVo = userInfoVoResult.getData();
            Assert.notNull(userInfoVo, "用户不能为空.");
            //  获取主播信息：
            albumInfoIndex.setAnnouncerName(userInfoVo.getNickname());
        }, threadPoolExecutor);

        //  获取分类信息
        CompletableFuture<Void> categoryCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            //  远程调用
            Result<BaseCategoryView> categoryViewResult = categoryFeignClient.getCategoryView(albumInfo.getCategory3Id());
            Assert.notNull(categoryViewResult, "远程调用分类返回值不能为空.");
            BaseCategoryView categoryView = categoryViewResult.getData();
            Assert.notNull(categoryView, "分类不能为空.");
            albumInfoIndex.setCategory1Id(categoryView.getCategory1Id());
            albumInfoIndex.setCategory2Id(categoryView.getCategory2Id());
            albumInfoIndex.setCategory3Id(categoryView.getCategory3Id());
        }, threadPoolExecutor);

        //  专辑统计：要么从数据库中获取.
        albumInfoIndex.setPlayStatNum(new Random().nextInt(100000));
        albumInfoIndex.setBuyStatNum(new Random().nextInt(10000));
        albumInfoIndex.setCommentStatNum(new Random().nextInt(10000000));
        albumInfoIndex.setSubscribeStatNum(new Random().nextInt(10000));

        CompletableFuture<Void> attrCompletableFuture = CompletableFuture.runAsync(() -> {
            //  专辑属性值
            Result<List<AlbumAttributeValue>> albumAttributeValueResult = albumInfoFeignClient.findAlbumAttributeValue(albumId);
            Assert.notNull(albumAttributeValueResult, "远程调用属性返回值不能为空.");
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueResult.getData();
            Assert.notNull(albumAttributeValueList, "属性不能为空.");
            //  循环遍历.
            List<AttributeValueIndex> attributeValueIndexList = albumAttributeValueList.stream().map(albumAttributeValue -> {
                //  创建对象
                AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                //  属性拷贝
                BeanUtils.copyProperties(albumAttributeValue, attributeValueIndex);
                //  返回数据
                return attributeValueIndex;
            }).collect(Collectors.toList());
            //  设置属性
            albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
        }, threadPoolExecutor);
        //  多任务组合：
        CompletableFuture.allOf(
                albumInfoCompletableFuture,
                categoryCompletableFuture,
                userInfoCompletableFuture,
                attrCompletableFuture
        ).join();

        //  保存数据到es中
        albumIndexRepository.save(albumInfoIndex);
        //  给提词库赋值.
        SuggestIndex suggestIndexTitle = new SuggestIndex();
        //  赋值： 专辑的标题
        suggestIndexTitle.setId(UUID.randomUUID().toString().replaceAll("-",""));
        suggestIndexTitle.setTitle(albumInfoIndex.getAlbumTitle());
        suggestIndexTitle.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumTitle()}));
        suggestIndexTitle.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumTitle())}));
        suggestIndexTitle.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumTitle())}));
        //  保存数据
        suggestIndexRepository.save(suggestIndexTitle);

        SuggestIndex suggestIndexIntro = new SuggestIndex();
        //  赋值： 专辑的简介
        suggestIndexIntro.setId(UUID.randomUUID().toString().replaceAll("-",""));
        suggestIndexIntro.setTitle(albumInfoIndex.getAlbumIntro());
        suggestIndexIntro.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumIntro()}));
        suggestIndexIntro.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumIntro())}));
        suggestIndexIntro.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumIntro())}));
        //  保存数据
        suggestIndexRepository.save(suggestIndexIntro);

        //  将专辑id 添加到布隆过滤器中.
        RBloomFilter<Object> bloomFilter = this.redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        //  添加数据
        bloomFilter.add(albumId);


    }


}
